package indi.tutubrother.rocktmq.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author tutubrother
 * @description
 * @date 2020/5/5 15:39
 */
public abstract class BaseOrderlyConsumer extends BaseConsumer {

    private static final Logger logger = LoggerFactory.getLogger(BaseConsumer.class);

    protected BaseOrderlyConsumer(String topic, String consumerGroup) {
        super(topic, consumerGroup);
    }

    private MessageListenerOrderly messageListenerOrderly = (msgList, context) -> {
        MessageExt messageExt = msgList.get(0);
        logger.info("{} receive msg: topic={}, tag={}, msgId={}, body={}",
                Thread.currentThread().getName(), messageExt.getTopic(), messageExt.getTags(),
                messageExt.getMsgId(), messageExt.getBody());
        boolean result = consumerAction.execute(new String(messageExt.getBody()));
        return result ? ConsumeOrderlyStatus.SUCCESS : ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
    };

    @Override
    public void start() throws MQClientException {
        init();
        check();
        DefaultMQPushConsumer consumer = initConsumer();
        consumer.registerMessageListener(messageListenerOrderly);
        consumer.start();
        logger.info("consumer(topic={}, tag={}, group={} started!)",
                this.topic, this.tag, this.consumerGroup);
    }
}
